package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;
import com.aotain.storm.TupleHelpers;

/**
 * 检测IP作为源的统计
 * @ClassName: AbnStatisBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2016年4月28日 下午1:33:32 
 *
 */
public class AbnReverseStatisBolt extends AbstractBolt{

	private static final long serialVersionUID = 1L;

	
	String dip = null;
	String dport = null;
	String sip = null;
	String sport = null;
	String dateStr_d = null;
	String dateStr_h = null;
	String dateStr_hour = null;
	String sourceAreaId = null;
	String areaName = null;
	String sourceAreaName = null;
	String idcno = null;
	String sourceCountry = null;
	Long triplePV = null;
	String abnormaltype = null;
	String pvKey = null;

	Map<String,Long> pvMap = new HashMap<String,Long>();

	

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(TupleHelpers.isTickTuple(tuple)) {
				emitCountingData(collector);  	
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println(" AbnStatisBolt exception"+e.getMessage());
			Logger.getRootLogger().error("AbnStatisBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException(e);
		}	
	}

	private void emitCountingData(OutputCollector collector) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);   


		SimpleDateFormat sdfm = new SimpleDateFormat("mm");
		SimpleDateFormat sdfh = new SimpleDateFormat("yyyyMMddHH00");
		SimpleDateFormat sdfh_ = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdfd = new SimpleDateFormat("yyyyMMdd");
		Date date = new Date();
		String dateStrM = sdfm.format(date);
		String dateStrH = sdfh.format(date);
		String dateStrH_ = sdfh_.format(date);
		String dateStrD = sdfd.format(date);
		int nMin5 = (Integer.parseInt(dateStrM)/5)*5;
		int nMin30 = (Integer.parseInt(dateStrM)/30)*30;
		String sMin5 = String.format("%02d", nMin5);
		String sMin30 = String.format("%02d", nMin30);
		String dateStr5 = dateStrH_+sMin5;
		String dateStr30 = dateStrH_+sMin30;
		String rowKeyAbnSession5Min = null;
		String rowKeyAbnSession30Min = null;
		String rowKeyAbnSessionH = null;
		String rowKeyAbnSessionD = null;

		for(Entry<String,Long> entry : pvMap.entrySet()) {
			String key  = entry.getKey();
			long value = entry.getValue().longValue();
			String[] splits3 = key.split("_",-1);
			rowKeyAbnSession5Min = splits3[0] + "_" + dateStr5;
			rowKeyAbnSession30Min = splits3[0] + "_" + dateStr30;
			rowKeyAbnSessionH = splits3[0] + "_" + dateStrH;
			rowKeyAbnSessionD = splits3[0] + "_" + dateStrD;
			hbaseInstance.incerment("SDS_SESSION_STAT_5MIN", rowKeyAbnSession5Min, "cf:OUTABNPV", value);
			hbaseInstance.incerment("SDS_SESSION_STAT_30MIN", rowKeyAbnSession30Min, "cf:OUTABNPV", value);
			hbaseInstance.incerment("SDS_SESSION_STAT_HOUR", rowKeyAbnSessionH, "cf:OUTABNPV", value);
			hbaseInstance.incerment("SDS_SESSION_STAT_DAY", rowKeyAbnSessionD, "cf:OUTABNPV", value);
		}
		pvMap.clear();
	}

	private void countInLocal(Tuple tuple) {

		if(tuple != null ) {  
			//			System.err.println("..........AbnStatisBolt......... " + System.currentTimeMillis());
			dip = tuple.getStringByField("dip");
			sip = tuple.getStringByField("sip");
			idcno = tuple.getStringByField("idcno");
			dateStr_d = tuple.getStringByField("dateStr_d"); 
//			dateStr_h = tuple.getStringByField("dateStr_h");
//			dateStr_hour = tuple.getStringByField("dateStr_hour");
//			sourceAreaId = tuple.getStringByField("sourceAreaId");
//			areaName = tuple.getStringByField("areaName");
//			sourceAreaName = tuple.getStringByField("sourceAreaName");
			triplePV = tuple.getLongByField("pv");
//			abnormaltype = tuple.getStringByField("abntype");
//			sourceCountry = tuple.getStringByField("sourcecountry");

			pvKey = sip + "_" +dateStr_d;
			long pv = triplePV;
			if(pvMap.get(pvKey) != null) {
				pv = pvMap.get(pvKey) + pv;
			}
			pvMap.put(pvKey, pv);
		}
	}

	@Override
	public void cleanup() {
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300);
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
	}

}
